/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.openness.crawler.frontier;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.util.Bytes;

import com.openness.crawler.hbase.HBaseUtil;
import com.openness.crawler.hbase.Row;
import com.openness.crawler.url.WebURL;
import com.openness.crawler.util.Constants;
import com.sleepycat.je.DatabaseException;

/**
 * This class maintains the list of pages which are assigned to crawlers but are
 * not yet processed. It is used for resuming a previous crawl.
 * 
 */
public class InProcessPagesDB extends CrawlerQueue {

	public InProcessPagesDB() throws DatabaseException {

	}

	@Override
	public List<WebURL> get(int max) throws IOException {
		List<WebURL> urls = new ArrayList<WebURL>();

		List<Row> rows = null;

		rows = HBaseUtil.getFirstRows(Constants.HBASE_TABLE_INPROCESSPAGESDB,
				max);

		if (rows != null && rows.size() > 0) {
			for (Row row : rows) {
				WebURL url = new WebURL();

				url.setURL(Bytes.toString(row
						.getQualifierValue(Constants.HBASE_TABLE_WEBURL_QUALIFIER_URL)));
				url.setDocid(Bytes.toLong(row
						.getQualifierValue(Constants.HBASE_TABLE_WEBURL_QUALIFIER_DOCID)));
				url.setParentDocid(Bytes.toLong(row
						.getQualifierValue(Constants.HBASE_TABLE_WEBURL_QUALIFIER_PARENTDOCID)));
				url.setParentUrl(Bytes.toString(row
						.getQualifierValue(Constants.HBASE_TABLE_WEBURL_QUALIFIER_PARENTURL)));
				url.setDepth(Bytes.toInt(row
						.getQualifierValue(Constants.HBASE_TABLE_WEBURL_QUALIFIER_DEPTH)));
				url.setPath(Bytes.toString(row
						.getQualifierValue(Constants.HBASE_TABLE_WEBURL_QUALIFIER_PATH)));
				url.setAnchor(Bytes.toString(row
						.getQualifierValue(Constants.HBASE_TABLE_WEBURL_QUALIFIER_ANCHOR)));
				url.setPriority(Bytes.toInt(row
						.getQualifierValue(Constants.HBASE_TABLE_WEBURL_QUALIFIER_PRIORITY)));

				urls.add(url);
			}
		}

		return urls;
	}

	@Override
	public boolean removeURL(WebURL webUrl) {
		try {
			HBaseUtil.deleteRow(Constants.HBASE_TABLE_INPROCESSPAGESDB,
					getRowKey(webUrl));

			return true;
		} catch (IOException e) {
		}

		return false;
	}

	@Override
	public void delete(List<WebURL> urls) throws IOException {
		List<byte[]> rows = null;

		if (urls != null && urls.size() > 0) {
			rows = new ArrayList<byte[]>(urls.size());

			for (WebURL url : urls) {
				rows.add(getRowKey(url));
			}
		}

		HBaseUtil.deleteRows(Constants.HBASE_TABLE_INPROCESSPAGESDB, rows);
	}

	@Override
	public void put(WebURL url) throws IOException {
		Row row = new Row(getRowKey(url));

		row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_URL,
				Bytes.toBytes(url.getURL()));
		row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_DOCID,
				Bytes.toBytes(url.getDocid()));
		row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_PARENTDOCID,
				Bytes.toBytes(url.getParentDocid()));
		row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_PARENTURL,
				Bytes.toBytes(url.getParentUrl()));
		row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_DEPTH,
				Bytes.toBytes(url.getDepth()));
		row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_DOMAIN,
				Bytes.toBytes(url.getDomain()));
		row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_SUBDOMAIN,
				Bytes.toBytes(url.getSubDomain()));
		row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_PATH,
				Bytes.toBytes(url.getPath()));
		row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_ANCHOR,
				Bytes.toBytes(url.getAnchor()));
		row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_PRIORITY,
				Bytes.toBytes(url.getPriority()));

		HBaseUtil.putRows(Constants.HBASE_TABLE_INPROCESSPAGESDB, row);
	}

	public void puts(List<WebURL> urls) throws IOException {
		if (urls == null || urls.size() == 0) {
			return;
		}

		List<Row> rows = new ArrayList<Row>();

		for (WebURL url : urls) {
			Row row = new Row(getRowKey(url));

			row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_URL,
					Bytes.toBytes(url.getURL()));
			row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_DOCID,
					Bytes.toBytes(url.getDocid()));
			row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_PARENTDOCID,
					Bytes.toBytes(url.getParentDocid()));
			row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_PARENTURL,
					Bytes.toBytes(url.getParentUrl()));
			row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_DEPTH,
					Bytes.toBytes(url.getDepth()));
			row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_DOMAIN,
					Bytes.toBytes(url.getDomain()));
			row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_SUBDOMAIN,
					Bytes.toBytes(url.getSubDomain()));
			row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_PATH,
					Bytes.toBytes(url.getPath()));
			row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_ANCHOR,
					Bytes.toBytes(url.getAnchor()));
			row.add(Constants.HBASE_TABLE_WEBURL_QUALIFIER_PRIORITY,
					Bytes.toBytes(url.getPriority()));

			rows.add(row);
		}

		HBaseUtil.putRows(Constants.HBASE_TABLE_INPROCESSPAGESDB,
				rows.toArray(new Row[rows.size()]));
	}

	@Override
	public long getLength() {
		try {
			return HBaseUtil.getRowCount(
					Constants.HBASE_TABLE_INPROCESSPAGESDB, null, null);
		} catch (Throwable e) {
		}

		return -1;
	}

}
